from typing import Any, Dict, List, Optional, Callable, Tuple

from ..core.graph import FlowGraph

from api.dao import create_run, update_run  # 适配新的 dao 入口便利函数


class BaseRunner:
    """
    通用的运行器基类，封装标准的运行生命周期与持久化逻辑。
    具体引擎只需实现差异化部分（如 engine_name、日志前缀），即可复用完整的执行流程。
    
    主要功能：
    - 标准化的流程执行生命周期
    - 外部/内部节点执行路由
    - 运行状态持久化（数据库 + Redis）
    - 断点续跑支持
    - 超时控制与异常处理
    """
    # 面向对象属性声明（类型注解 + 注释），用于提升可读性与一致性
    graph: FlowGraph  # 流程图对象，包含节点和边的定义
    max_steps: int  # 最大执行步数，防止无限循环
    checkpoint_file: Optional[str]  # 检查点文件路径，用于断点续跑
    resume: bool  # 是否从检查点恢复执行
    node_timeout: Optional[float]  # 单个节点执行超时时间（秒），None 表示无超时
    parallelism: int  # 并行度设置，当前版本暂未使用
    stop_requested: bool  # 是否收到停止请求，用于优雅终止流程
    _flow_id: Optional[int]  # 流程 ID，用于数据库关联
    _run_id: Optional[int]  # 运行记录 ID，用于数据库关联
    _exec_map: Dict[str, Any]  # 节点执行映射配置缓存（内部/外部执行路由）
    _node_seq: int  # 节点序号计数器，用于外部调用的唯一记录构造
    trace: List[str]  # 统一的执行轨迹（日志消息），用于观测与调试

    def __init__(
        self,
        graph: FlowGraph,
        max_steps: int = 1000,
        checkpoint_file: Optional[str] = None,
        resume: bool = False,
        node_timeout: Optional[float] = None,
        parallelism: int = 1,
        flow_id: Optional[int] = None,
        run_id: Optional[int] = None,
    ):
        """
        初始化运行器基类。
        
        Args:
            graph: 流程图对象，包含节点和边的定义
            max_steps: 最大执行步数，防止无限循环，默认 1000
            checkpoint_file: 检查点文件路径，用于断点续跑，可选
            resume: 是否从检查点恢复执行，默认 False
            node_timeout: 单个节点执行超时时间（秒），None 表示无超时
            parallelism: 并行度设置，当前版本暂未使用，默认 1
            flow_id: 流程 ID，用于数据库关联，可选
            run_id: 运行记录 ID，用于数据库关联，可选
        """
        self.graph = graph
        self.max_steps = max_steps
        self.checkpoint_file = checkpoint_file
        self.resume = resume
        self.node_timeout = node_timeout
        self.parallelism = max(1, int(parallelism))
        self.stop_requested = False
        # 持久化相关
        self._flow_id = flow_id
        self._run_id = run_id
        # 外部执行映射
        self._exec_map: Dict[str, Any] = {}
        self._node_seq = 0
        # 统一的执行轨迹（trace）
        self.trace: List[str] = []

    # ====== 需子类提供的差异化能力 ======
    def engine_name(self) -> str:
        """
        返回引擎名称，用于日志标识和数据库记录。
        
        Returns:
            str: 引擎名称，如 'FlowRunner'、'LangGraphRunner' 等
            
        Raises:
            NotImplementedError: 子类必须实现此方法
        """
        raise NotImplementedError

    def log(self, message: str) -> None:
        """
        记录日志消息，默认实现为打印到控制台并添加到执行轨迹。
        子类可覆盖此方法以添加引擎特定的日志前缀或格式。
        
        Args:
            message: 要记录的日志消息
        """
        print(message)
        self.trace.append(message)

    # ====== 持久化辅助 ======
    def _persist_start(self, context: Dict[str, Any]) -> None:
        """
        执行开始时的持久化操作，包括创建运行记录和写入初始上下文。
        
        功能：
        - 如果 run_id 为空且 flow_id 存在，则创建新的运行记录
        - 更新运行状态为 'running'
        - 将初始上下文写入 Redis 供外部观测（已调整为在 runId 写入到 context 后执行）
        
        Args:
            context: 初始执行上下文
            
        Note:
            所有异常都会被捕获并记录为警告，不会中断执行流程
        """
        try:
            # 兼容旧结构：直接使用函数式 API，避免循环依赖
            if self._run_id is None and self._flow_id is not None:
                run_dict = create_run(
                    flow_id=self._flow_id,
                    status='pending',
                    engine=self.engine_name(),
                    max_steps=self.max_steps,
                    timeout=self.node_timeout,
                    parallelism=None,
                    context_json=context,
                )
                self._run_id = run_dict.get('id')
            if self._run_id is not None:
                update_run(self._run_id, status='running')
            # 确保将 runId 写入到上下文；若持久化失败则生成临时 runId
            try:
                context.setdefault('flow', {})
                if self._run_id is not None:
                    context['flow']['runId'] = str(self._run_id)
                else:
                    existing = str((context.get('flow') or {}).get('runId') or '').strip()
                    if not existing:
                        import uuid
                        tmp_id = f"tmp:{uuid.uuid4().hex}"
                        context['flow']['runId'] = tmp_id
                        self.log(f"[WARN] Persist start: runId unavailable; using temporary runId={tmp_id}")
            except Exception as ce:
                self.log(f"[WARN] Set context flow.runId failed: {ce}")
        except Exception as e:
            self.log(f"[WARN] Persist start failed: {e}")
            # 持久化出现异常时也确保上下文中有临时 runId，避免后续报错
            try:
                context.setdefault('flow', {})
                import uuid
                tmp_id = f"tmp:{uuid.uuid4().hex}"
                context['flow']['runId'] = tmp_id
                self.log(f"[WARN] Using temporary runId={tmp_id} due to persist start error")
            except Exception:
                pass

    def _persist_finish(self, context: Dict[str, Any], visited_steps: int, status: str, error: Optional[str] = None, trace: Optional[List[str]] = None) -> None:
        """
        执行结束时的持久化操作，更新运行记录的最终状态和结果。
        
        行为说明：
        - 成功更新数据库后，会尝试将最终上下文写入 Redis（键：{run_id}_context）。
        - 当 Redis 写入失败时：记录 ERROR 日志、设置 self.stop_requested=True，并抛出 RuntimeError 中断流程。
        
        Args:
            context (Dict[str, Any]): 最终执行上下文
            visited_steps (int): 总访问步数
            status (str): 完成状态（completed/stopped/error）
            error (Optional[str]): 错误信息（仅在异常完成时存在）
            trace (Optional[List[str]]): 执行轨迹列表
        """
        try:
            # 兼容旧结构：直接使用函数式 API，避免循环依赖
            if self._run_id is not None:
                from datetime import datetime, timezone
                fields: Dict[str, Any] = {
                    'status': status,
                    'finished_at': datetime.now(timezone.utc).replace(tzinfo=None).isoformat(),
                    'trace_json': trace if trace is not None else [],
                    'context_json': context,
                    'visited_steps': visited_steps,
                }
                if error:
                    fields['error'] = error
                update_run(self._run_id, **fields)
                # 结束时也写入最终上下文到 Redis，便于外部消费 #{flowRecordId}_context
                try:
                    # 优先使用 context.flow.runId，确保与外部节点使用的 flowRunId 一致
                    run_id = None
                    try:
                        run_id = str((context.get('flow') or {}).get('runId') or '').strip()
                    except Exception:
                        pass
                    
                    # 如果 context.flow.runId 为空，回退到 self._run_id
                    if not run_id:
                        run_id = str(self._run_id or '').strip()
                    
                    if run_id:
                        from api.managers.redis_manager import REDIS_MANAGER
                        key = f"{run_id}_context"
                        REDIS_MANAGER.set_json(key, context)
                        self.log(f"[REDIS] Finish context written to key={key}")
                    else:
                        self.log(f"[WARN] Redis finish skipped: no valid run_id available")
                except Exception as re:
                    self.log(f"[ERROR] Redis write finish failed: {re}")
                    self.stop_requested = True
                    raise RuntimeError(f"Redis finish context write failed: {re}")
        except Exception as e:
            self.log(f"[WARN] Persist finish failed: {e}")

    def _persist_snapshot(self, context: Dict[str, Any]) -> None:
        """
        执行过程中的快照持久化，将当前上下文写入 Redis。
        
        功能：
        - 每执行完一个节点后调用，保证上下文实时更新
        - 使用键格式：{run_id}_context
        - 供下游系统或外部监控实时获取执行状态
        
        Args:
            context (Dict[str, Any]): 当前执行上下文
            
        Note:
            如果 run_id 为空则跳过；当 Redis 写入失败时，记录 ERROR、设置 self.stop_requested=True，并抛出 RuntimeError 以中断执行。
        """
        try:
            # 优先使用 context.flow.runId，确保与外部节点使用的 flowRunId 一致
            run_id = None
            try:
                run_id = str((context.get('flow') or {}).get('runId') or '').strip()
            except Exception:
                pass
            
            # 如果 context.flow.runId 为空，回退到 self._run_id
            if not run_id:
                run_id = str(self._run_id or '').strip()
            
            if not run_id:
                self.log(f"[WARN] Redis snapshot skipped: no valid run_id available")
                return
                
            from api.managers.redis_manager import REDIS_MANAGER
            key = f"{run_id}_context"
            REDIS_MANAGER.set_json(key, context)
            self.log(f"[REDIS] Snapshot written to key={key}")
        except Exception as re:
            self.log(f"[ERROR] Redis write snapshot failed: {re}")
            self.stop_requested = True
            raise RuntimeError(f"Redis snapshot write failed: {re}")

    # ====== 检查点（断点续跑） ======
    def _save_checkpoint(self, context: Dict[str, Any], queue: List[str], visited_steps: int) -> None:
        """
        保存检查点到文件，支持断点续跑功能。
        
        功能：
        - 将当前执行状态（上下文、队列、步数、轨迹）序列化到 JSON 文件
        - 在每个节点执行完成后调用
        - 仅在设置了 checkpoint_file 时生效
        
        Args:
            context: 当前执行上下文
            queue: 待执行节点队列
            visited_steps: 已访问步数
            
        Note:
            异常会被捕获并记录为警告，不会中断执行
        """
        if not self.checkpoint_file:
            return
        try:
            import json
            payload = {
                'context': context,
                'queue': queue,
                'visited_steps': visited_steps,
                'trace': self.trace,
            }
            with open(self.checkpoint_file, 'w', encoding='utf-8') as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)
        except Exception as e:
            self.log(f"[WARN] Failed to save checkpoint: {e}")

    def _load_checkpoint(self) -> Optional[Dict[str, Any]]:
        """
        从检查点文件加载之前的执行状态。
        
        功能：
        - 读取并解析检查点 JSON 文件
        - 恢复上下文、队列、步数和执行轨迹
        - 仅在启用 resume 且文件存在时调用
        
        Returns:
            Optional[Dict[str, Any]]: 检查点数据，包含 context、queue、visited_steps、trace
                                    如果文件不存在或解析失败则返回 None
                                    
        Note:
            异常会被捕获并记录为警告，返回 None
        """
        if not self.checkpoint_file:
            return None
        try:
            import os
            import json
            if os.path.exists(self.checkpoint_file):
                with open(self.checkpoint_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
        except Exception as e:
            self.log(f"[WARN] Failed to load checkpoint: {e}")
        return None

    # ====== 外部/内部执行与路由 ======
    def _load_exec_map(self) -> None:
        """
        加载节点执行映射配置，决定节点是外部执行还是内部执行。
        
        功能：
        - 从 config/node_exec_map.json 读取配置
        - 标准化配置格式，支持 mode、endpoint、timeout、headers 等字段
        - 配置缓存，避免重复读取
        
        配置格式示例：
        {
            "http_request": {
                "mode": "external",
                "endpoint": "http://localhost:8080/execute",
                "timeout": 30,
                "headers": {"Authorization": "Bearer token"}
            },
            "condition": "internal"
        }
        
        Note:
            异常会被捕获，回退到全部内部执行模式
        """
        if self._exec_map:
            return
        try:
            import json, os
            cfg_path = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', 'config', 'node_exec_map.json'))
            with open(cfg_path, 'r', encoding='utf-8') as f:
                raw = json.load(f)
            normalized: Dict[str, Any] = {}
            for k, v in raw.items():
                if isinstance(v, dict):
                    mode = (v.get('mode') or v.get('type') or 'internal')
                    normalized[k] = {
                        'mode': mode,
                        'endpoint': v.get('endpoint'),
                        'timeout': v.get('timeout'),
                        'headers': v.get('headers') or {},
                    }
                else:
                    normalized[k] = {'mode': str(v)}
            self._exec_map = normalized
        except Exception as e:
            self.log(f"[WARN] Load node exec map failed: {e}. Fallback to internal.")
            self._exec_map = {}

    def _is_external(self, node_type: str) -> bool:
        """
        判断指定节点类型是否需要外部执行。
        
        Args:
            node_type: 节点类型，如 'http_request'、'condition' 等
            
        Returns:
            bool: True 表示外部执行，False 表示内部执行
            
        Note:
            会自动加载执行映射配置，默认为内部执行
        """
        self._load_exec_map()
        entry = self._exec_map.get(node_type) or {}
        return (entry.get('mode') or 'internal') == 'external'

    def _next_from_status(self, node_id: str, status: Optional[Any]) -> List[str]:
        """
        根据节点执行状态确定下一步要执行的节点列表。
        
        功能：
        - 支持单个 handle 字符串或 handle 列表
        - 优先匹配指定的 source_handle
        - 如果没有匹配的边，回退到默认边（source_handle 为 None 或 'default'）
        
        Args:
            node_id: 当前节点 ID
            status: 节点执行状态，可以是字符串、列表或其他类型
                   - 字符串：单个 source_handle
                   - 列表/元组：多个 source_handle
                   - 其他：使用默认边
                   
        Returns:
            List[str]: 下一步要执行的节点 ID 列表
            
        Examples:
            - status="success" -> 查找 source_handle="success" 的边
            - status=["success", "warning"] -> 查找这两个 handle 的所有边
            - status=None -> 使用默认边
        """
        # 接受单个 handle 字符串或列表；若缺失则回退到默认边
        if isinstance(status, (list, tuple)):
            handles = [str(h) for h in status if h is not None]
            targets: List[str] = []
            for h in handles:
                targets.extend([e.target for e in self.graph.get_outgoing_edges(node_id) if e.source_handle == h])
            if targets:
                return targets
        elif isinstance(status, str) and status:
            targets = [e.target for e in self.graph.get_outgoing_edges(node_id) if e.source_handle == status]
            if targets:
                return targets
        return [e.target for e in self.graph.get_outgoing_edges(node_id) if e.source_handle in (None, 'default')]

    def external_execute(self, node: Any, context: Dict[str, Any], log_cb: Optional[Callable[[str], None]] = None) -> List[str]:
        """
        执行外部节点，通过 HTTP POST 调用外部服务。
        
        功能：
        - 构造标准化的请求载荷，包含流程元数据和节点信息
        - 支持多种 URL 来源：节点配置中的 externalUrl/url 或映射配置中的 endpoint
        - 解析响应中的路由信息（sourceHandle/nodeStatus）
        - 异常或缺失路由时自动回退到内部执行
        - 记录完整的请求/响应到上下文供调试
        
        Args:
            node: 要执行的节点对象
            context: 当前执行上下文
            log_cb: 可选的日志回调函数，默认使用 self.log
            
        Returns:
            List[str]: 下一步要执行的节点 ID 列表
            
        请求载荷格式：
        {
            "flowRunId": "运行记录ID",
            "flowMeta": {
                "flowId": "流程ID", 
                "flowConfig": "流程配置（JSON字符串）"
            },
            "nodeRunId": "节点运行记录ID（唯一）",
            "nodeMeta": {
                "nodeId": "节点ID",
                "nodeConfig": "节点配置（JSON字符串）",
                "nodeType": "节点类型"
            }
        }
        
        响应解析：
        - 优先从 response.data.sourceHandle 获取路由
        - 回退到 response.sourceHandle 或 response.nodeStatus
        - 缺失路由信息时回退到内部执行
        
        Note:
            所有外部调用的请求和响应都会记录到 context[f"node_{node.id}_external"]
        """
        from urllib import request as urlreq, error as urlerr
        import json, uuid
        self._node_seq += 1
        # 优先使用 context.flow.runId；若不存在则使用数据库 run_id；如果仍为空则生成临时 runId 并写回 context
        flow_run_id = None
        try:
            flow_run_id = str((context.get('flow') or {}).get('runId') or '').strip()
        except Exception:
            flow_run_id = ''
        if not flow_run_id:
            flow_run_id = str(self._run_id or '').strip()
        if not flow_run_id:
            flow_run_id = f"tmp:{uuid.uuid4().hex}"
            try:
                context.setdefault('flow', {})
                context['flow']['runId'] = flow_run_id
            except Exception:
                pass
        node_record_id = f"{flow_run_id}:{node.id}:{self._node_seq}:{uuid.uuid4().hex[:8]}"
        payload = {
            "flowRunId": flow_run_id,
            "flowMeta": {
                "flowId": str(self._flow_id or ''),
                "flowConfig": json.dumps(self.graph.flow, ensure_ascii=False),
            },
            "nodeRunId": node_record_id,
            "nodeMeta": {
                "nodeId": node.id,
                "nodeConfig": json.dumps(node.data or {}, ensure_ascii=False),
                "nodeType": node.type,
            },
        }
        # 打印外部调用的请求参数（原样 JSON）用于排查问题
        logger = log_cb or self.log
        try:
            logger(f"[EXT] Node {node.id} request payload: {json.dumps(payload, ensure_ascii=False)}")
        except Exception:
            pass
        data = node.data or {}
        mapping = self._exec_map.get(node.type) or {}
        endpoint = data.get('externalUrl') or data.get('url') or mapping.get('endpoint')
        logger = log_cb or self.log
        if not endpoint:
            logger(f"[EXT] Node {node.id} ({node.type}) external URL missing; fallback to internal execution.")
            context[f"node_{node.id}_external"] = {"request": payload, "response": None, "error": "missing_url", "fallback": "internal"}
            # fallback to internal execution when endpoint missing
            return self.execute_internal_with_timeout(node, self.graph, context, logger)
        try:
            body = json.dumps(payload).encode('utf-8')
            headers = dict(mapping.get('headers') or {})
            headers["Content-Type"] = "application/json"
            req = urlreq.Request(url=endpoint, method='POST', headers=headers, data=body)
            timeout = int(self.node_timeout or data.get('timeout') or mapping.get('timeout') or 30)
            logger(f"[EXT] POST {endpoint} for node {node.id}")
            with urlreq.urlopen(req, timeout=timeout) as resp:
                resp_text = resp.read().decode('utf-8', errors='replace')
                try:
                    resp_json = json.loads(resp_text)
                except Exception:
                    resp_json = {"raw": resp_text}
                context[f"node_{node.id}_external"] = {"request": payload, "response": resp_json}
                data_block = resp_json.get('data') if isinstance(resp_json, dict) else None
                node_status = None
                end_time = None
                source_handles = None
                if isinstance(data_block, dict):
                    node_status = data_block.get('nodeStatus')
                    end_time = data_block.get('endTime')
                    source_handles = data_block.get('sourceHandle')
                else:
                    if isinstance(resp_json, dict):
                        node_status = resp_json.get('nodeStatus')
                        end_time = resp_json.get('endTime')
                        source_handles = resp_json.get('sourceHandle')
                node.output = {"external": resp_json, "endTime": end_time}
                targets = self._next_from_status(node.id, source_handles or node_status)
                if not targets:
                    logger(f"[EXT] Node {node.id} external response missing routing; fallback to internal execution.")
                    # mark fallback for observability
                    try:
                        entry = context.get(f"node_{node.id}_external", {})
                        if isinstance(entry, dict):
                            entry["fallback"] = "internal"
                            context[f"node_{node.id}_external"] = entry
                    except Exception:
                        pass
                    return self.execute_internal_with_timeout(node, self.graph, context, logger)
                return targets
        except urlerr.URLError as e:
            logger(f"[EXT] Node {node.id} external call error: {e}; fallback to internal execution.")
            context[f"node_{node.id}_external"] = {"request": payload, "response": None, "error": str(e), "fallback": "internal"}
            return self.execute_internal_with_timeout(node, self.graph, context, logger)

    def execute_internal_with_timeout(self, node: Any, graph: FlowGraph, context: Dict[str, Any], log_cb: Optional[Callable[[str], None]] = None) -> List[str]:
        """
        执行内部节点，支持超时控制。
        
        功能：
        - 如果设置了 node_timeout，使用线程池执行并等待结果
        - 超时时记录错误并返回空列表（跳过该节点）
        - 无超时设置时直接调用节点的 execute 方法
        
        Args:
            node: 要执行的节点对象
            graph: 流程图对象
            context: 当前执行上下文
            log_cb: 可选的日志回调函数，默认使用 self.log
            
        Returns:
            List[str]: 下一步要执行的节点 ID 列表，超时时返回空列表
            
        Note:
            超时异常会被捕获并记录，不会中断整个流程执行
        """
        logger = log_cb or self.log
        if self.node_timeout and self.node_timeout > 0:
            from concurrent.futures import ThreadPoolExecutor, TimeoutError
            with ThreadPoolExecutor(max_workers=1) as ex:
                fut = ex.submit(node.execute, graph, context, logger)
                try:
                    return fut.result(timeout=self.node_timeout)
                except TimeoutError:
                    logger(f"[ERROR] Node {node.id} execution timed out after {self.node_timeout}s. Skipping.")
                    return []
        else:
            return node.execute(graph, context, logger)
        return []

    # ====== 差异化点的统一封装 ======
    def should_continue(self, queue: List[str], visited_steps: int) -> bool:
        """
        判断是否应该继续执行流程。
        
        功能：
        - 检查是否收到停止请求
        - 检查是否超过最大步数限制（防止无限循环）
        - 检查队列是否为空
        
        Args:
            queue: 待执行节点队列
            visited_steps: 已访问步数
            
        Returns:
            bool: True 表示继续执行，False 表示停止执行
            
        停止条件：
        1. 收到外部停止请求（self.stop_requested = True）
        2. 执行步数超过 max_steps 限制
        3. 待执行队列为空
        
        Note:
            子类可以覆盖此方法实现自定义的继续策略
        """
        if self.stop_requested:
            self.log('[INFO] Stop requested. Stopping execution.')
            return False
        if visited_steps > self.max_steps:
            self.log('[ERROR] Max steps exceeded; possible loop detected. Stopping.')
            return False
        return bool(queue)

    def execute_node(self, node: Any, context: Dict[str, Any]) -> List[str]:
        """
        统一的节点执行入口，根据节点类型选择外部或内部执行。
        
        功能：
        - 根据节点类型查询执行映射配置
        - 外部节点调用 external_execute
        - 内部节点调用 execute_internal_with_timeout
        
        Args:
            node: 要执行的节点对象
            context: 当前执行上下文
            
        Returns:
            List[str]: 下一步要执行的节点 ID 列表
            
        Note:
            这是节点执行的统一分派点，子类通常不需要覆盖此方法
        """
        if self._is_external(node.type):
            return self.external_execute(node, context)
        else:
            return self.execute_internal_with_timeout(node, self.graph, context, self.log)

    # ====== 标准运行生命周期 ======
    def _init_flow_context(self, context: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        """
        初始化流程运行上下文的基础信息（flow 元数据与完整配置）。
        - 设置 flow.id、flow.runId(占位)、flow.startedAt
        - 回填缺失的 flow.id（优先级：self._flow_id > graph.flow.id > 'local'）
        - 注入完整的流程配置到 context.flow.config（字典）
        
        Args:
            context: 可选的初始上下文
        
        Returns:
            Dict[str, Any]: 规范化后的上下文对象
        """
        ctx = context or {}
        try:
            import datetime as _dt
            started_at = _dt.datetime.now(_dt.timezone.utc).isoformat().replace('+00:00', 'Z')
            # 注入系统环境参数（供编辑器与节点引用）：env
            try:
                from api.utils import DEFAULT_ENGINE
                k_enabled = False
                try:
                    import os
                    k_enabled = (os.getenv('KAFKA_ENABLED') or 'false').lower() in ('1','true','yes','on')
                except Exception:
                    pass
                ctx.setdefault('env', {})
                ctx['env'].setdefault('kafkaEnabled', bool(k_enabled))
                ctx['env'].setdefault('defaultEngine', DEFAULT_ENGINE)
                ctx['env'].setdefault('timezone', 'UTC')
            except Exception:
                pass
            # 补充可配置的全局变量容器（globals.vars）
            try:
                ctx.setdefault('globals', {})
                g = ctx['globals']
                if isinstance(g, dict):
                    g.setdefault('vars', {})
            except Exception:
                pass
            flow_meta = {
                'id': str(self._flow_id or ''),
                'runId': None,
                'startedAt': started_at,
            }
            if not isinstance(ctx.get('flow'), dict):
                ctx['flow'] = flow_meta
            else:
                ctx['flow'].setdefault('id', str(self._flow_id or ''))
                ctx['flow'].setdefault('runId', None)
                ctx['flow'].setdefault('startedAt', started_at)
            # 补充 flow.id 与流程配置
            try:
                ctx.setdefault('flow', {})
                _fid = str(ctx['flow'].get('id') or '').strip()
                if not _fid:
                    if self._flow_id is not None:
                        ctx['flow']['id'] = str(self._flow_id)
                    elif isinstance(getattr(self.graph, 'flow', None), dict) and (getattr(self.graph, 'flow', {}).get('id') is not None):
                        ctx['flow']['id'] = str(getattr(self.graph, 'flow', {}).get('id'))
                    else:
                        ctx['flow']['id'] = 'local'
                if isinstance(getattr(self.graph, 'flow', None), dict):
                    ctx['flow'].setdefault('config', self.graph.flow)
            except Exception:
                pass
        except Exception:
            # 忽略元数据初始化异常，确保流程能够继续执行
            pass
        return ctx

    def _ensure_run_id_in_context(self, context: Dict[str, Any]) -> None:
        """
        确保上下文中存在可用的 flow.runId。
        - 优先使用持久化产生的 self._run_id
        - 如果不可用则生成临时 runId 并写入上下文
        
        Args:
            context: 当前执行上下文
        """
        try:
            context.setdefault('flow', {})
            run_in_ctx = str((context.get('flow') or {}).get('runId') or '').strip()
            if not run_in_ctx:
                if self._run_id is not None:
                    context['flow']['runId'] = str(self._run_id)
                else:
                    import uuid
                    tmp_id = f"tmp:{uuid.uuid4().hex}"
                    context['flow']['runId'] = tmp_id
                    self.log(f"[WARN] No persisted runId; using temporary runId={tmp_id}")
        except Exception:
            pass

    def _write_start_context_to_redis(self, context: Dict[str, Any]) -> None:
        """
        将初始上下文写入 Redis（键：{run_id}_context）。
        当写入失败时记录 ERROR、设置停止标志并抛出 RuntimeError 中断执行。
        
        Args:
            context: 当前执行上下文
        """
        try:
            run_id = None
            try:
                run_id = str((context.get('flow') or {}).get('runId') or '').strip()
            except Exception:
                pass
            if not run_id:
                run_id = str(self._run_id or '').strip()
            if run_id:
                from api.managers.redis_manager import REDIS_MANAGER
                key = f"{run_id}_context"
                REDIS_MANAGER.set_json(key, context)
                self.log(f"[REDIS] Start context written to key={key}")
            else:
                self.log(f"[WARN] Redis start skipped: no valid run_id available")
        except Exception as re:
            self.log(f"[ERROR] Redis write start failed: {re}")
            self.stop_requested = True
            raise RuntimeError(f"Redis start context write failed: {re}")

    def _prepare_initial_queue(self, context: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str], int]:
        """
        准备初始队列与步数（支持断点续跑）。
        - 当 resume 且存在检查点文件时，加载检查点恢复 context/queue/visited_steps/trace
        - 否则，定位开始节点并以其为初始队列
        
        Args:
            context: 当前执行上下文
        
        Returns:
            Tuple[Dict[str, Any], List[str], int]: (可能更新后的上下文, 初始队列, 已访问步数)
        """
        visited_steps = 0
        queue: List[str] = []
        if self.resume and self.checkpoint_file:
            loaded = self._load_checkpoint()
            if loaded:
                context = loaded.get('context', context)
                queue = loaded.get('queue', [])
                visited_steps = int(loaded.get('visited_steps', 0))
                self.trace = list(loaded.get('trace', []))
            else:
                start = self.graph.find_start_node()
                if not start:
                    raise RuntimeError('Start node not found')
                queue = [start.id]
        else:
            start = self.graph.find_start_node()
            if not start:
                raise RuntimeError('Start node not found')
            queue = [start.id]
        return context, queue, visited_steps

    def run(self, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        标准运行流程：准备队列 -> 节点循环 -> 快照/检查点 -> 完成持久化。
        子类无需重复编写主循环，只需覆盖必要的差异化钩子。
        
        Note:
            在初始化阶段，会将初始上下文写入 Redis；若写入失败，将记录 ERROR、设置停止标志并抛出 RuntimeError 中断流程。
        """
        # 1) 初始化上下文（flow 元信息与完整配置）
        context = self._init_flow_context(context)
        # 2) 持久化并确保上下文包含 runId
        self._persist_start(context)
        self._ensure_run_id_in_context(context)
        # 3) 将初始上下文写入 Redis
        self._write_start_context_to_redis(context)
        # 4) 准备初始队列与步数（可选断点续跑）
        context, queue, visited_steps = self._prepare_initial_queue(context)
        try:
            while self.should_continue(queue, visited_steps):
                current_id = queue.pop(0)
                node = self.graph.nodes_by_id.get(current_id)
                if not node:
                    self.log(f"[WARN] Node {current_id} not found.")
                    self._save_checkpoint(context, queue, visited_steps)
                    continue
                # 更新当前节点上下文
                try:
                    context['current'] = {'nodeId': getattr(node, 'id', '?')}
                except Exception:
                    pass
                import datetime
                ts = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                self.log(f"[NODE] Enter {getattr(node, 'id', '?')} ({getattr(node, 'type', '?')}) at {ts}")
                try:
                    node.input = self._interpolate_struct(getattr(node, 'input', {}), context)
                    node.data = self._interpolate_struct(getattr(node, 'data', {}), context)
                except Exception as e:
                    self.log(f"[WARN] Interpolate for node {getattr(node, 'id', '?')} failed: {e}")
                next_nodes = self.execute_node(node, context)
                # 记录节点执行结果
                try:
                    nodes_ctx = context.setdefault('nodes', {})
                    nodes_ctx[node.id] = {
                        'id': node.id,
                        'type': getattr(node, 'type', None),
                        'input': getattr(node, 'input', None),
                        'output': getattr(node, 'output', None),
                        'data': getattr(node, 'data', None),
                        'ts': ts,
                    }
                except Exception as e:
                    self.log(f"[WARN] Record node output failed: {e}")
                # 快照写入 Redis
                self._persist_snapshot(context)
                visited_steps += 1
                for nxt in next_nodes:
                    if nxt:
                        queue.append(nxt)
                self._save_checkpoint(context, queue, visited_steps)
            status = 'completed' if not self.stop_requested else 'stopped'
            self._persist_finish(context, visited_steps, status, trace=self.trace)
            return {'context': context, 'trace': self.trace}
        except Exception as e:
            self._persist_finish(context, visited_steps, 'error', error=str(e), trace=self.trace)
            raise

    def _interpolate_value(self, val: Any, ctx: Dict[str, Any]) -> Any:
        """
        在单个值中插值解析 ${...} 形式的占位符。
        
        规则：
        - 如果值是字符串且包含占位符，则按上下文进行替换；
        - 如果整个字符串仅是一个占位符且解析结果为非字符串，直接返回其“原始类型”的值；
        
        Args:
            val (Any): 待解析的值（字符串/数字/布尔/对象等）
            ctx (Dict[str, Any]): 执行上下文，用于路径解析
        
        Returns:
            Any: 解析后的值；若无法解析或非字符串则原样返回
        """
        try:
            if isinstance(val, str):
                import re
                pattern = re.compile(r"\$\{([^}]+)\}")
                matches = list(pattern.finditer(val))
                if not matches:
                    return val
                # 如果整个字符串恰好是一个占位符，则直接返回解析到的“原始类型”值
                if len(matches) == 1 and matches[0].group(0) == val:
                    resolved = self._resolve_path(ctx, matches[0].group(1))
                    return resolved
                # 否则进行字符串级替换（将 None 替换为空字符串）
                def repl(m):
                    resolved = self._resolve_path(ctx, m.group(1))
                    return "" if resolved is None else str(resolved)
                return pattern.sub(repl, val)
            return val
        except Exception:
            return val

    def _resolve_path(self, ctx: Dict[str, Any], path: str) -> Any:
        """
        从上下文中解析点分路径（例如：nodes.n1.output.text 或 globals.input.userId）。
        
        Args:
            ctx (Dict[str, Any]): 执行上下文对象
            path (str): 点分路径字符串
        
        Returns:
            Any: 解析到的值；若路径不存在或解析失败，返回 None
        """
        try:
            parts = [p for p in str(path).split('.') if p]
            cur: Any = ctx
            for p in parts:
                if isinstance(cur, dict) and p in cur:
                    cur = cur[p]
                else:
                    return None
            return cur
        except Exception:
            return None

    def _interpolate_struct(self, obj: Any, ctx: Dict[str, Any]) -> Any:
        """
        递归解析字典/列表/字符串结构中的占位符，返回同结构的解析结果。
        支持传统的 ${...} 插值和 JsonLogic 表达式。
        
        Args:
            obj (Any): 待解析的对象（dict/list/str/其他）
            ctx (Dict[str, Any]): 执行上下文，用于占位符解析
        
        Returns:
            Any: 解析后的对象，类型与输入保持一致
        """
        if isinstance(obj, dict):
            # 检查是否为 JsonLogic 表达式
            if self._is_jsonlogic_expression(obj):
                return self._evaluate_jsonlogic_expression(obj, ctx)
            # 普通字典递归处理
            return {k: self._interpolate_struct(v, ctx) for k, v in obj.items()}
        if isinstance(obj, list):
            return [self._interpolate_struct(v, ctx) for v in obj]
        return self._interpolate_value(obj, ctx)
    
    def _is_jsonlogic_expression(self, obj: Any) -> bool:
        """
        检查对象是否为 JsonLogic 表达式
        
        Args:
            obj: 待检查的对象
            
        Returns:
            是否为 JsonLogic 表达式
        """
        return (isinstance(obj, dict) and 
                len(obj) == 1 and 
                "$logic" in obj)
    
    def _evaluate_jsonlogic_expression(self, expression: Dict[str, Any], ctx: Dict[str, Any]) -> Any:
        """
        计算 JsonLogic 表达式
        
        Args:
            expression: JsonLogic 表达式字典
            ctx: 执行上下文
            
        Returns:
            计算结果
        """
        try:
            from ..utils.jsonlogic_resolver import JsonLogicResolver
            resolver = JsonLogicResolver(ctx)
            return resolver.resolve(expression)
        except Exception as e:
            self.log(f"[WARN] JsonLogic expression evaluation failed: {e}")
            return expression